package sttp.tapir.server.tests

import cats.implicits._
import org.scalatest.concurrent.Eventually
import org.scalatest.concurrent.Eventually.eventually
import org.scalatest.matchers.should.Matchers._
import sttp.client4._
import sttp.monad.MonadError
import sttp.monad.syntax._
import sttp.model.StatusCode
import sttp.tapir._
import sttp.tapir.server.interceptor.CustomiseInterceptors
import sttp.tapir.server.interceptor.metrics.MetricsRequestInterceptor
import sttp.tapir.server.interceptor.reject.DefaultRejectHandler
import sttp.tapir.server.metrics.{EndpointMetric, Metric}
import sttp.tapir.server.tests.ServerMetricsTest._
import sttp.tapir.tests.Basic.{in_input_stream_out_input_stream, in_json_out_json, in_root_path}
import sttp.tapir.tests.Test
import sttp.tapir.tests.TestUtil.inputStreamToByteArray

import java.io.{ByteArrayInputStream, InputStream}
import java.util.concurrent.atomic.AtomicInteger
import sttp.tapir.server.ServerEndpoint

class ServerMetricsTest[F[_], OPTIONS, ROUTE](
    createServerTest: CreateServerTest[F, Any, OPTIONS, ROUTE],
    serverInterpreter: TestServerInterpreter[F, Any, OPTIONS, ROUTE],
    supportsMetricsDecodeFailureCallbacks: Boolean = true
)(implicit m: MonadError[F]) {
  import createServerTest._
  import serverInterpreter._

  // increase the patience for `eventually` for slow CI tests
  implicit val patienceConfig: Eventually.PatienceConfig = Eventually.PatienceConfig(
    timeout = org.scalatest.time.Span(15, org.scalatest.time.Seconds),
    interval = org.scalatest.time.Span(150, org.scalatest.time.Millis)
  )

  def tests(): List[Test] = List(
    {
      val reqCounter = newRequestCounter[F]
      val resCounter = newResponseCounter[F]
      val metrics = new MetricsRequestInterceptor[F](List(reqCounter, resCounter), Seq.empty)

      testServer(
        in_json_out_json.name("metrics"),
        interceptors = (ci: CustomiseInterceptors[F, OPTIONS]) => ci.metricsInterceptor(metrics)
      )(f => (if (f.fruit == "apple") Right(f) else Left(())).unit) { (backend, baseUri) =>
        basicRequest // onDecodeSuccess path
          .post(uri"$baseUri/api/echo")
          .body("""{"fruit":"apple","amount":1}""")
          .send(backend)
          .map { r =>
            r.body shouldBe Right("""{"fruit":"apple","amount":1}""")
            reqCounter.metric.value.get() shouldBe 1
            eventually { // the response metric is invoked *after* the body is sent, so we might have to wait
              resCounter.metric.value.get() shouldBe 1
            }
          } >> basicRequest // onDecodeFailure path
          .post(uri"$baseUri/api/echo")
          .body("""{"invalid":"body",}""")
          .send(backend)
          .map { _ =>
            reqCounter.metric.value.get() shouldBe 2
            eventually {
              resCounter.metric.value.get() shouldBe 2
            }
          }
      }
    }, {
      val resCounter = newResponseCounter[F]
      val metrics = new MetricsRequestInterceptor[F](List(resCounter), Seq.empty)

      testServer(
        in_input_stream_out_input_stream.name("metrics"),
        interceptors = (ci: CustomiseInterceptors[F, OPTIONS]) => ci.metricsInterceptor(metrics)
      )(is => blockingResult((new ByteArrayInputStream(inputStreamToByteArray(is)): InputStream).asRight[Unit])) { (backend, baseUri) =>
        basicRequest
          .post(uri"$baseUri/api/echo")
          .body("okoń")
          .send(backend)
          .map { r =>
            r.body shouldBe Right("okoń")
            eventually {
              resCounter.metric.value.get() shouldBe 1
            }
          }
      }
    }, {
      val resCounter = newResponseCounter[F]
      val metrics = new MetricsRequestInterceptor[F](List(resCounter), Seq.empty)

      testServer(in_root_path.name("metrics"), interceptors = (ci: CustomiseInterceptors[F, OPTIONS]) => ci.metricsInterceptor(metrics))(
        _ => ().asRight[Unit].unit
      ) { (backend, baseUri) =>
        basicRequest
          .get(uri"$baseUri")
          .send(backend)
          .map { r =>
            r.body shouldBe Right("")
            eventually {
              resCounter.metric.value.get() shouldBe 1
            }
          }
      }
    }, {
      val reqCounter = newRequestCounter[F]
      val resCounter = newResponseCounter[F]
      val metrics = new MetricsRequestInterceptor[F](List(reqCounter, resCounter), Seq.empty)

      testServer(
        in_root_path.name("metrics on exception"),
        interceptors = (ci: CustomiseInterceptors[F, OPTIONS]) => ci.metricsInterceptor(metrics)
      ) { _ =>
        Thread.sleep(100)
        throw new RuntimeException("Ups")
      } { (backend, baseUri) =>
        basicRequest
          .get(uri"$baseUri")
          .send(backend)
          .map { _ =>
            eventually {
              reqCounter.metric.value.get() shouldBe 1
              resCounter.metric.value.get() shouldBe 1
            }
          }
      }
    }, {
      // Test that metrics are collected when an exception is thrown
      val exceptionCounter = newExceptionCounter[F]
      val reqCounter = newRequestCounter[F]
      val metrics = new MetricsRequestInterceptor[F](List(reqCounter, exceptionCounter), Seq.empty)

      testServer(
        endpoint.get.in("test-path").out(stringBody),
        "metrics on exception from server logic",
        interceptors = (ci: CustomiseInterceptors[F, OPTIONS]) => ci.metricsInterceptor(metrics).exceptionHandler(None)
      ) { _ =>
        throw new RuntimeException("Test exception")
      } { (backend, baseUri) =>
        basicRequest
          .get(uri"$baseUri/test-path")
          .send(backend)
          .map { _ =>
            eventually {
              reqCounter.metric.value.get() shouldBe 1
              exceptionCounter.metric.value.get() shouldBe 1
            }
          }
          .recover {
            case e: RuntimeException if e.getMessage == "Test exception" =>
              // ok, in some interpreters (e.g. test stubs) the exception falls through
              succeed
          }
      }
    }
  ) ++ metricsCallbacksTests()

  def metricsCallbacksTests(): List[Test] =
    // these tests have to be disable for servers which perform their own path matching: then, when the path doesn't match, the
    // Tapir code isn't called at all.
    if (supportsMetricsDecodeFailureCallbacks)
      List(
        {
          // Test that metrics are collected for responses generated by interceptors (e.g., 404 from RejectInterceptor)
          val interceptorResponseCounter = newInterceptorResponseCounter[F]
          val metrics = new MetricsRequestInterceptor[F](List(interceptorResponseCounter), Seq.empty)

          testServer(
            "metrics on request handler response (404 from RejectInterceptor)",
            route(
              // multiple endpoints so that the reject interceptor is not disabled
              List[ServerEndpoint[Any, F]](
                endpoint.get.in("test-path1").out(stringBody).serverLogic((_: Unit) => pureResult("ok1".asRight[Unit])),
                endpoint.get.in("test-path2").out(stringBody).serverLogic((_: Unit) => pureResult("ok2".asRight[Unit]))
              ),
              // when all endpoints fail to decode the request, the reject interceptor returns 404
              interceptors =
                (ci: CustomiseInterceptors[F, OPTIONS]) => ci.metricsInterceptor(metrics).rejectHandler(DefaultRejectHandler.orNotFound[F])
            )
          ) { (backend, baseUri) =>
            basicRequest
              .get(uri"$baseUri/unknown-path")
              .send(backend)
              .map { response =>
                response.code shouldBe StatusCode.NotFound
                eventually {
                  interceptorResponseCounter.metric.value.get() shouldBe 1
                }
              }
          }
        }, {
          // Test that metrics are collected when all endpoints fail to decode the request
          val decodeFailureCounter = newDecodeFailureCounter[F]
          val metrics = new MetricsRequestInterceptor[F](List(decodeFailureCounter), Seq.empty)

          testServer(
            "metrics on decode failure",
            route(
              // multiple endpoints so that the reject interceptor is not disabled
              List[ServerEndpoint[Any, F]](
                endpoint.get.in("test-path").out(stringBody).serverLogic((_: Unit) => pureResult("ok1".asRight[Unit])),
                endpoint.get.in("test-path2").out(stringBody).serverLogic((_: Unit) => pureResult("ok2".asRight[Unit]))
              ),
              // when all endpoints fail to decode the request, the reject interceptor doesn't do anything
              interceptors =
                (ci: CustomiseInterceptors[F, OPTIONS]) => ci.metricsInterceptor(metrics).rejectHandler(DefaultRejectHandler.apply[F])
            )
          ) { (backend, baseUri) =>
            basicRequest
              .get(uri"$baseUri/unknown-path")
              .send(backend)
              .map { _ =>
                eventually {
                  decodeFailureCounter.metric.value.get() shouldBe 1
                }
              }
          }
        }
      )
    else Nil
}

object ServerMetricsTest {
  case class Counter(value: AtomicInteger = new AtomicInteger(0)) {
    def ++(): Unit = value.incrementAndGet()
  }

  def newRequestCounter[F[_]]: Metric[F, Counter] =
    Metric[F, Counter](Counter(), onRequest = { (_, c, m) => m.unit(EndpointMetric().onEndpointRequest { _ => m.eval(c.++()) }) })

  def newResponseCounter[F[_]]: Metric[F, Counter] =
    Metric[F, Counter](Counter(), onRequest = { (_, c, m) => m.unit(EndpointMetric().onResponseBody { (_, _) => m.eval(c.++()) }) })

  def newInterceptorResponseCounter[F[_]]: Metric[F, Counter] =
    Metric[F, Counter](
      Counter(),
      onRequest = { (_, c, m) => m.unit(EndpointMetric().onInterceptorResponse { _ => m.eval(c.++()) }) }
    )

  def newDecodeFailureCounter[F[_]]: Metric[F, Counter] =
    Metric[F, Counter](
      Counter(),
      onRequest = { (_, c, m) => m.unit(EndpointMetric().onDecodeFailure { () => m.eval(c.++()) }) }
    )

  def newExceptionCounter[F[_]]: Metric[F, Counter] =
    Metric[F, Counter](
      Counter(),
      onRequest = { (_, c, m) => m.unit(EndpointMetric().onException { (_, _) => m.eval(c.++()) }) }
    )
}
